package com.yifeng.repo.controller.kafka.message.consume;

import com.yifeng.repo.controller.kafka.message.handler.KafkaHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.common.TopicPartition;

import java.util.Collection;

/**
 * Created by daibing on 2024/5/2.
 */
@Slf4j
public class LoggingConsumerRebalanceListener implements ConsumerRebalanceListener {
    /**
     * rebalance发生之前调用
     */
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        log.info("onPartitions Revoked: {}", KafkaHelper.buildPartitionLog(partitions));
    }

    /**
     * rebalance发生之后调用
     */
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        log.info("onPartitions Assigned: {}", KafkaHelper.buildPartitionLog(partitions));
    }

}
